package com.kai.mqtt.message.service.impl;

import com.kai.mqtt.data.PageResult;
import com.kai.mqtt.data.request.PageRetainMessageRequest;
import com.kai.mqtt.data.response.RetainMessageResponse;
import com.kai.mqtt.message.RetainMessage;
import com.kai.mqtt.message.service.IRetainMessageService;
import com.kai.mqtt.util.TopicUtil;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author weijk
 * @since 2024/6/10
 */
@Service
public class RetainMessageServiceImpl implements IRetainMessageService {

    private static final Map<String, RetainMessage> MAP = new ConcurrentHashMap<>();

    @Override
    public void put(String topic, RetainMessage retainMessage) {
        MAP.put(topic, retainMessage);
    }

    @Override
    public void remove(String topic) {
        MAP.remove(topic);
    }

    @Override
    public List<RetainMessage> search(String topicFilter) {
        List<RetainMessage> retainMessages = new ArrayList<>();
        MAP.forEach((topic, message) -> {
            if (TopicUtil.match(topicFilter, topic)) {
                retainMessages.add(message);
            }
        });

        return retainMessages;
    }

    /**
     * 分页查询保留消息列表
     *
     * @param request 参数
     * @return 保留消息列表
     */
    @Override
    public PageResult<RetainMessageResponse> pageRetain(PageRetainMessageRequest request) {
        List<RetainMessage> retainMessages = MAP.values().stream()
                .filter(retain -> !StringUtils.hasText(request.getTopic()) || Objects.equals(request.getTopic(), retain.getTopic()))
                .toList();

        List<RetainMessageResponse> responses = retainMessages.stream()
                .skip((long) request.getPageSize() * (request.getPageIndex() - 1))
                .limit(request.getPageSize())
                .map(retainMessage -> {
                    RetainMessageResponse response = new RetainMessageResponse();
                    BeanUtils.copyProperties(retainMessage, response);

                    return response.setMessage(new String(retainMessage.getMessageBytes()));
                }).toList();

        return PageResult.success(responses, request.getPageIndex(), request.getPageSize(), retainMessages.size());
    }

    @Override
    public Boolean deleteRetain(String topic) {
        this.remove(topic);

        return Boolean.TRUE;
    }
}
